package org.breathe.dynamic.thread.pool.core.executor;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.github.dadiyang.equator.Equator;
import com.github.dadiyang.equator.FieldInfo;
import com.github.dadiyang.equator.GetterBaseEquator;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.breathe.dynamic.thread.pool.core.context.BreatheDtpContext;
import org.breathe.dynamic.thread.pool.core.context.BreatheDtpContextHolder;
import org.breathe.dynamic.thread.pool.core.convert.ExecutorConvert;
import org.breathe.dynamic.thread.pool.core.handler.NotifierHandler;
import org.breathe.dynamic.thread.pool.core.notify.alarm.AlarmLimiter;
import org.breathe.dynamic.thread.pool.core.notify.NotifyHelper;
import org.breathe.dynamic.thread.pool.core.reject.RejectHandlerGetter;
import org.breathe.dynamic.thread.pool.core.spring.ExecutorWrapper;
import org.breathe.dynamic.thread.pool.core.spring.ThreadPoolCreator;
import org.breathe.dynamic.thread.pool.core.thread.BreatheDtpExecutor;
import org.breathe.dynamic.thread.pool.core.thread.ThreadPoolBuilder;
import org.breathe.thread.pool.common.constant.BreatheDtpConstant;
import org.breathe.thread.pool.common.entity.BreatheDtpMainProperties;
import org.breathe.thread.pool.common.entity.BreatheDtpProperties;
import org.breathe.thread.pool.common.entity.NotifyItem;
import org.breathe.thread.pool.common.entity.ThreadPoolExecutorProperties;
import org.breathe.thread.pool.common.enums.NotifyTypeEnum;
import org.breathe.thread.pool.common.enums.QueueTypeEnum;
import org.breathe.thread.pool.common.exception.DtpException;
import org.breathe.thread.pool.common.queue.VariableLinkedBlockingQueue;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;

/**
 * 线程池注册
 * @author breathe
 * @createTime: 2025-04-08
 */
@Slf4j
public class ThreadPoolExecutorRegistry implements InitializingBean {

    /**
     * 发送消息线程池
     */
    private static final ExecutorService NOTIFY_EXECUTOR = ThreadPoolCreator.createCommonWithTtl("breathe-dtp-notify");
    /**
     * 维护所有自动和手动注册的breathe动态线程池
     */
    private static final Map<String, BreatheDtpExecutor> DTP_REGISTRY = new ConcurrentHashMap<>();
    /**
     * 维护所有自动和手动注册的JUC线程池
     */
    private static final Map<String, ExecutorWrapper> COMMON_REGISTRY = new ConcurrentHashMap<>();
    private static BreatheDtpProperties breatheDtpProperties;

    /**
     * 比较差异
     */
    private static final Equator EQUATOR = new GetterBaseEquator();
    /**
     * 获得存放线程池的Map
     * @return 返回注册的线程池
     */
    public static Map<String, BreatheDtpExecutor> getDtpRegistry() {
        return DTP_REGISTRY;
    }

    /**
     *
     * @param breatheDtpExecutor 自定义的动态线程池
     */
    public static void put(BreatheDtpExecutor breatheDtpExecutor) {
        log.info("thread-pool register -> [source: {}, executor: {}]", breatheDtpExecutor.getThreadPoolName(), breatheDtpExecutor);
    }

    /**
     * 动态线程池注册
     */
    public static void registerDynamic(BreatheDtpExecutor executor, String source) {
        log.info("thread-pool register -> [source: {}, executor: {}]", source, executor);
        DTP_REGISTRY.putIfAbsent(executor.getThreadPoolName(), executor);
    }

    /**
     * JUC线程池注册
     */
    public static void registerCommon(ExecutorWrapper wrapper, String source) {
        log.info("thread-pool register -> [source: {}, executor: {}]", source, wrapper.getThreadPoolName());
        COMMON_REGISTRY.putIfAbsent(wrapper.getThreadPoolName(), wrapper);
    }
    /**
     * 获取线程池
     */
    public static BreatheDtpExecutor get(String threadPoolName) {
        final BreatheDtpExecutor executor = DTP_REGISTRY.get(threadPoolName);
        if (Objects.isNull(executor)) {
            log.error("can't find dtp that name is {}", threadPoolName);
            throw new DtpException("can't find dtp that name is" + threadPoolName);
        }
        return executor;
    }

    /**
     * 当监听到配置改变的时候刷新
     */
    public static void refresh(BreatheDtpProperties properties) {
        if (Objects.isNull(properties) || CollUtil.isEmpty(properties.getExecutors())) {
            log.warn("breathe dynamic thread pool refresh, empty properties");
            return ;
        }
        properties.getExecutors().forEach(each -> {
            if (StrUtil.isBlank(each.getThreadPoolName())) {
                log.warn("breathe dynamic thread pool refresh, thread-pool-name must not be empty");
                return ;
            }
            final BreatheDtpExecutor breatheDtpExecutor = DTP_REGISTRY.get(each.getThreadPoolName());
            if (Objects.isNull(breatheDtpExecutor)) {
                log.warn("breathe dynamic thread pool refresh, can't find this executor, name: {}", each.getThreadPoolName());
                return ;
            }
            refresh(breatheDtpExecutor, each);
        });
    }

    public static void refresh(BreatheDtpExecutor executor, ThreadPoolExecutorProperties properties) {
        BreatheDtpMainProperties oldProperties = ExecutorConvert.convert(executor);
        doRefresh(executor, properties);
        BreatheDtpMainProperties newProperties = ExecutorConvert.convert(executor);
        if (oldProperties.equals(newProperties)) {
            log.warn("breathe dynamic thread pool [{}] has no properties changed.", executor.getThreadPoolName());
            return ;
        }
        List<FieldInfo> diffFields = EQUATOR.getDiffFields(oldProperties, newProperties);
        List<String> diffKeys = diffFields.stream()
                .map(FieldInfo::getFieldName)
                .toList();
        log.info("breathe dynamic thread pool [{}] refreshed end, changed keys: {}, corePoolSize: [{}], maxPoolSize: [{}], " +
                        "queueType: [{}], queueCapacity: [{}], keepAliveTime: [{}], rejectedType: [{}], " +
                        "allowsCoreThreadTimeOut: [{}]",
                executor.getThreadPoolName(),
                diffKeys,
                String.format(BreatheDtpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.getCorePoolSize(), newProperties.getCorePoolSize()),
                String.format(BreatheDtpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.getMaxPoolSize(), newProperties.getMaxPoolSize()),
                String.format(BreatheDtpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.getQueueType(), newProperties.getQueueType()),
                String.format(BreatheDtpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.getQueueCapacity(), newProperties.getQueueCapacity()),
                String.format(BreatheDtpConstant.PROPERTIES_CHANGE_TIME_SHOW_STYLE, oldProperties.getKeepAliveTime(), newProperties.getKeepAliveTime()),
                String.format(BreatheDtpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.getRejectType(), newProperties.getRejectType()),
                String.format(BreatheDtpConstant.PROPERTIES_CHANGE_SHOW_STYLE, oldProperties.isAllowCoreThreadTimeOut(), newProperties.isAllowCoreThreadTimeOut())
        );
        final NotifyItem notifyItem = NotifyHelper.getNotifyItem(executor, NotifyTypeEnum.CHANGE);
        boolean ifNotice = CollUtil.isNotEmpty(breatheDtpProperties.getPlatforms()) && Objects.nonNull(notifyItem) && notifyItem.isEnabled();
        if (!ifNotice) {
            return ;
        }
        BreatheDtpContext context = BreatheDtpContext.builder()
                .breatheDtpExecutor(executor)
                .platforms(breatheDtpProperties.getPlatforms())
                .notifyItem(notifyItem)
                .build();
        BreatheDtpContextHolder.set(context);
        // 让一个线程去报警
        NOTIFY_EXECUTOR.execute(() -> NotifierHandler.getInstance().sendNotice(oldProperties, diffKeys));
    }
    public static void doRefresh(BreatheDtpExecutor breatheDtpExecutor, ThreadPoolExecutorProperties properties) {
        if (properties.getMaximumPoolSize() >= 0) {
            breatheDtpExecutor.setMaximumPoolSize(properties.getMaximumPoolSize());
        }
        if (properties.getCorePoolSize() > breatheDtpExecutor.getMaximumPoolSize()) {
            throw new IllegalArgumentException();
        }
        if (properties.getCorePoolSize() >= 0) {
            breatheDtpExecutor.setCorePoolSize(properties.getCorePoolSize());
        }
        if (properties.getKeepAliveTime() > 0 && properties.getTimeUnit() != null) {
            breatheDtpExecutor.setKeepAliveTime(properties.getKeepAliveTime(), properties.getTimeUnit());
        }
        breatheDtpExecutor.allowCoreThreadTimeOut(properties.isAllowCoreThreadTimeOut());
        String originRejectHandlerName = breatheDtpExecutor.getRejectHandlerName();
        if (StrUtil.isNotBlank(properties.getRejectedHandlerType()) && !originRejectHandlerName.contains(properties.getRejectedHandlerType())) {
            breatheDtpExecutor.setRejectedExecutionHandler(RejectHandlerGetter.getProxy(properties.getRejectedHandlerType()));
            breatheDtpExecutor.setRejectHandlerName(properties.getRejectedHandlerType());
        }
        if (properties.getQueueCapacity() > 0 && Objects.equals(properties.getQueueType(), QueueTypeEnum.VARIABLE_LINKED_BLOCKING_QUEUE.getName())) {
            final BlockingQueue<Runnable> blockingQueue = breatheDtpExecutor.getQueue();
            if (blockingQueue instanceof VariableLinkedBlockingQueue<Runnable> variableLinkedBlockingQueue) {
                variableLinkedBlockingQueue.setCapacity(properties.getQueueCapacity());
            } else {
                log.error("breathe dynamic thread pool refresh, the blockingQueue capacity can't be reset, executor's name:{}, queueType: {}", breatheDtpExecutor.getThreadPoolName(), breatheDtpExecutor.getQueueName());
            }
        }
        if (CollUtil.isEmpty(properties.getNotifyItems())) {
            breatheDtpExecutor.setNotifyItems(NotifyItem.getDefaultNotifyItems());
            return ;
        }
        NotifyHelper.setExecutorNotifyItems(breatheDtpExecutor, breatheDtpProperties, properties);
        breatheDtpExecutor.setNotifyItems(properties.getNotifyItems());
    }

    /**
     * 获取所有的动态线程池名称
     */
    public static List<String> listAllBreatheDtpNames() {
        return Lists.newArrayList(DTP_REGISTRY.keySet());
    }

    @Autowired
    public void setBreatheDtpProperties(BreatheDtpProperties breatheDtpProperties) {
        ThreadPoolExecutorRegistry.breatheDtpProperties = breatheDtpProperties;
    }
    /**
     *
     * @throws Exception 抛出异常
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        //如果配置中没有线程池，直接结束
        if (CollectionUtils.isEmpty(breatheDtpProperties.getExecutors())) {
            return ;
        }
        //遍历所有配置的线程池
        breatheDtpProperties.getExecutors().forEach(each -> {
            final BreatheDtpExecutor executor = ThreadPoolBuilder.newBuilder()
                    .corePoolSize(each.getCorePoolSize())
                    .maximumPoolSize(each.getMaximumPoolSize())
                    .keepAliveTime(each.getKeepAliveTime())
                    .workQueue(each.getQueueType(), each.getQueueCapacity(), each.getFair())
                    .rejectedExecutionHandler(each.getRejectedHandlerType())
                    .threadFactory(each.getThreadNamePrefix())
                    .allowCoreThreadTimeOut(each.isAllowCoreThreadTimeOut())
                    .threadPoolName(each.getThreadPoolName())
                    .notifyItems(each.getNotifyItems())
                    .buildDynamic();
            registerDynamic(executor, "configuration");
        });
        DTP_REGISTRY.forEach((key, value) -> {
            NotifyHelper.fillNotifyItems(breatheDtpProperties.getPlatforms(), value.getNotifyItems());
            value.getNotifyItems().forEach(each -> AlarmLimiter.initAlarmLimiter(key, each));
        });
    }
}
